package com.zues.rabbit_springboot_consumer.listener;

import com.alibaba.fastjson2.JSONArray;
import com.rabbitmq.client.Channel;
import com.zues.rabbit_springboot_consumer.entity.Student;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.TimeUnit;

/**
 * @ClassName MyMessageListener
 * @Description TODO
 * @Author zeus
 * @Date 6/22/2025 8:19 下午
 * @Version 1.0
 **/
@Component
@Slf4j
public class MyMessageListener {

    public static final String EXCHANGE_DIRECT = "exchange.direct.order";
    public static final String ROUTING_KEY = "order";
    public static final String QUEUE_NAME = "queue.order";

    //写法一：监听 + 在RabbitMQ 服务器上创建交换机、队列
//    @RabbitListener(bindings = @QueueBinding(value = @Queue(value = QUEUE_NAME, durable = "true"),
//            exchange = @Exchange(value = EXCHANGE_DIRECT),
//            key = {ROUTING_KEY}
//    ))

    // 写法二： 监听
//    @RabbitListener(queues = QUEUE_NAME)
    public void processMessage(String dataString, Message message, Channel channel) throws IOException {

        //DeliveryTag: 每条消息在队列中的唯一标识
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        try {
            //核心操作
            log.info("消费端接受到了消息： {}", dataString);

            // 核心操作成功，返回 ACK 信息
            channel.basicAck(deliveryTag, false);

        } catch (Exception e) {
            log.error("消费端处理消息失败：{}", e.getMessage());
            // 获取当前消息是否重复投递的
            Boolean redelivered = message.getMessageProperties().getRedelivered();

            /*
             * 核心操作失败： 返回 NACK 信息
             * requeue 参数： 控制消息是否重新放回队列
             *  true:   重新放回队列，broker 会重新投递这个消息
             *  false:  不重新放回队列，broker 会丢弃这个消息
             *
             */
            if (redelivered) {
                // 如果当前消息已经是重复投递的，说明此前已经重试过一次了，此处可以选择不同的操作。比如重试几次，或重试几次不成功后直接记录日志
                // 此处requeue设置为false，标识不重新放回队列
                channel.basicNack(deliveryTag, false, false);
            }
            //重试 如果当前消息是第一次投递，说明当前代码是第一次抛异常，尚未重试，所以requeue设置为true，表示重新放回队列
            channel.basicNack(deliveryTag, false, true);
        }


    }

    /**
     * 测试限流监听
     *
     * @param dataString
     * @param message
     * @param channel
     * @throws IOException
     */
//    @RabbitListener(queues = QUEUE_NAME)
    @RabbitListener(bindings = @QueueBinding(value = @Queue(value = QUEUE_NAME, durable = "true"),
            exchange = @Exchange(value = EXCHANGE_DIRECT),
            key = {ROUTING_KEY}
    ))
    public void processMessageTestPrefetch(String dataString, Message message, Channel channel) throws Exception {
        //核心操作
        log.info("消费端接受到了消息： {}", dataString);
        TimeUnit.SECONDS.sleep(1);
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
    }


    public static final String OBJ_EXCHANGE_DIRECT = "exchange.test.obj";
    public static final String OBJ_ROUTING_KEY = "routing.key.test.obj";
    public static final String OBJ_QUEUE_NAME = "queue.test.obj";

//    @RabbitListener(queues = OBJ_QUEUE_NAME)
    public void processObjMessage(String dataString, Message message, Channel channel) throws IOException {

        //DeliveryTag: 每条消息在队列中的唯一标识
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        try {
            //核心操作
            log.info("消费端接受到了消息： {}", dataString);
            if (Objects.nonNull(dataString) && !"".equals(dataString)) {
//                Student student = JSON.parseObject(dataString, Student.class);
//                System.out.println("student = " + student);
                List<Student> students = JSONArray.parseArray(dataString, Student.class);
                students.forEach(System.out::println);
            }

            // 核心操作成功，返回 ACK 信息
            channel.basicAck(deliveryTag, false);

        } catch (Exception e) {
            log.error("消费端处理消息失败：{}", e.getMessage());
            // 获取当前消息是否重复投递的
            Boolean redelivered = message.getMessageProperties().getRedelivered();

            /*
             * 核心操作失败： 返回 NACK 信息
             * requeue 参数： 控制消息是否重新放回队列
             *  true:   重新放回队列，broker 会重新投递这个消息
             *  false:  不重新放回队列，broker 会丢弃这个消息
             *
             */
            if (redelivered) {
                // 如果当前消息已经是重复投递的，说明此前已经重试过一次了，此处可以选择不同的操作。比如重试几次，或重试几次不成功后直接记录日志
                // 此处requeue设置为false，标识不重新放回队列
                channel.basicNack(deliveryTag, false, false);
            }
            //重试 如果当前消息是第一次投递，说明当前代码是第一次抛异常，尚未重试，所以requeue设置为true，表示重新放回队列
            channel.basicNack(deliveryTag, false, true);
        }


    }


    public static final String QUEUE_DELAY = "queue.delay.happy";

    /**
     * 测试延迟队列
     * @param dataString  消息内容
     * @param message     消息对象
     * @param channel     通道对象
     * @throws IOException  异常
     */
    @RabbitListener(queues = {QUEUE_DELAY})
    public void process(String dataString, Message message, Channel channel) throws IOException {
        log.info("[生产者]" + dataString);
        log.info("[消费者]" + new SimpleDateFormat("hh:mm:ss").format(new Date()));
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
    }

}
